package com.hzlj.position.config.common.dto.common;

import cn.hutool.core.thread.NamedThreadFactory;
import cn.hutool.core.thread.ThreadUtil;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.RateLimiter;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;


@Slf4j
public class LocateTimerTask implements TimerTask {
    private final static Map<String, Task> tasks = Maps.newConcurrentMap();
    private final static ExecutorService executor = ThreadUtil.newExecutor();
    private final static AtomicBoolean running = new AtomicBoolean(true);
    private static HashedWheelTimer timer;
    private Task task;
    private LinkedBlockingQueue<Object> taskQueue;


    public synchronized static void start(List<TaskConfig> taskConfigs,
                                          Consumer<Task> positionConsumer) {
        if (timer != null) {
            log.warn("不需要重新启动定位定时调度");
            return;
        }
        //timer
        timer = new HashedWheelTimer(new NamedThreadFactory("定位定时调度", true));
        //开启线程调用
        for (TaskConfig config : taskConfigs) {
            LinkedBlockingQueue<?> taskQueue = config.getTaskQueue();
            RateLimiter rateLimiter = config.getRateLimiter();
            //多线程运行
            ThreadUtil.execAsync(() -> {
                while (running.get()) {
                    Task taskCopy = null;
                    try {
                        Task task = (Task) taskQueue.take();
                        int size = taskQueue.size();
                        taskCopy = task;
                        rateLimiter.acquire();
                        log.debug("定位定时调度-运行:{},size={}", task, size);
                        executor.execute(() -> {
                            try {
                                positionConsumer.accept(task);
                            } catch (Exception e) {
                                log.error("定位定时调度-运行-失败:{},size={}", task, size, e);
                            }
                        });
                    } catch (Exception e) {
                        log.error("定位定时调度-运行-等待调度超时:{}", taskCopy);
                    }
                }
            });
        }
    }

    public static void cancel(Task task) {
        Task remove = tasks.remove(task.key());
        if (remove != null) {
            log.info("定位定时调度-移除:{}", task);
        }
    }

    public static void stop() {
        if (timer != null) {
            timer.stop();
        }
        running.set(false);
    }

    /**
     * 不存在的情况再新增
     */
    public synchronized static void addIfAbsent(Task task, LinkedBlockingQueue<Object> taskQueue) {
        if (tasks.containsKey(task.key())) {
            return;
        }
        add(task, taskQueue);
    }

    public synchronized static void add(Task task, LinkedBlockingQueue<Object> taskQueue) {
        if (timer == null) {
            log.warn("请先启动定位定时调度");
            return;
        }
        if (task.getLocateInterval() <= 0) {
            log.warn("定位间隔设置成不需要定位:task={}", task);
            return;
        }
        LocateTimerTask locateTimerTask = new LocateTimerTask();
        locateTimerTask.task = task;
        locateTimerTask.taskQueue = taskQueue;
        Task old = tasks.put(task.key(), task);

        //需要修改批次
        if (old != null) {
            task.setBatch(old.getBatch() + 1);
        }
        //开始调度
        timer.newTimeout(locateTimerTask, task.getDelay(), TimeUnit.SECONDS);
        log.info("定位定时调度-添加:task={}", task);
    }

    @Override
    public void run(Timeout timeout) {
        String taskKey = this.task.key();
        if (timeout.isCancelled() || !tasks.containsKey(taskKey)) {
            log.info("定位定时调度-已被移除:{}", task);
            timeout.cancel();
            return;
        }

        //如果批次不相等，则直接移除
        Task taskNew = tasks.get(taskKey);
        if (!taskNew.getBatch().equals(this.task.getBatch())) {
            log.info("定位定时调度-已被覆盖:new->{},old->{}", taskNew, this.task);
            timeout.cancel();
            return;
        }
        if (taskQueue == null) {
            log.info("定位定时调度-没有配置队列:task={}", this.task);
            return;
        }
        //添加任务
        boolean offer = taskQueue.offer(taskNew);
        if (!offer) {
            log.info("定位定时调度-队列已满,被忽略:{},size={}", taskNew, taskQueue.size());
            return;
        }

        //重新调度
        try {
            timer.newTimeout(this, task.getLocateInterval(), TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("定位定时调度-设置调度失败:{},size={}", taskNew, taskQueue.size(), e);
        }
    }


    @Data
    public static class Task {
        private Key key;
        //延迟的秒数
        private Integer delay;
        //秒数
        private Integer locateInterval;
        //批次，标记是不是同一个任务
        private Long batch;

        public Task(Key key) {
            this(key, 3600, 0);
        }


        public Task(Key key,
                    Integer locateInterval,
                    Integer delay) {
            this.key = key;
            this.locateInterval = locateInterval;
            this.batch = 1L;
            this.delay = delay == null ? 0 : delay;
        }

        public String key() {
            return key.getDeviceCode();
        }

        @Override
        public String toString() {
            return "Task{" + "key='" + key
                    + ", locateInterval=" + locateInterval
                    + ", delay=" + delay
                    + ", batch=" + batch
                    + '}';
        }

        @Data
        @AllArgsConstructor
        public static class Key {
            private String jgCode;
            private String jzId;
            private String deviceCode;
            private String channel;
        }
    }
}
